/********************************************************/
/*	Copyright (C) 2016 Gong Li Bin			*/
/*	Project:	MapReduce			*/
/*	Author:		gong_libin			*/
/*	Date:		2016_06_01			*/
/*	File:		MrAcceptTest.cpp		*/
/********************************************************/

#include "MrJob.h"

#define	MR_AT_PORT			(char*)"10011"
#define	MR_AT_ADDR			(char*)"127.0.0.1"

bool g_bPrint = false;

void MrAtHelp(char* pszProc);
void MrAtResult(char* pszUrl, char* pszKey);

int main(int argc, char* argv[])
{
	CMrUdp CUdp;
	CMrTime CTime;
	int iIndex = 0;
	int iReturn = 0;
	int iLength = 0;
	MRPROKV_S stOut;
	CMrPro CProSend;
	CMrPro CProRecv;
	UINT uiMax = 100;
	UINT uiCount = 0;
	char szLcl[] = "I am local";
	char szMr[] = "I am global";
	char szKey[MR_VALUE] = { 0 };
	UCHAR uszRecv[MR_PACKET] = { 0 };
	UCHAR uszSend[MR_PACKET] = { 0 };
	MRPROKV_S stIn[] = {
		{MR_MR_MR, strlen(szMr), (UCHAR*)szMr},
		{MR_MR_MAP, strlen("101"), (UCHAR*)"101"},
		{MR_MR_MAP, strlen("111"), (UCHAR*)"111"},
		{MR_MR_LCL, strlen(szLcl), (UCHAR*)szLcl},
		{MR_MR_MAP, strlen("102"), (UCHAR*)"102"},
		{MR_MR_MAP, strlen("112"), (UCHAR*)"112"},
		{MR_MR_LCL, strlen(szLcl), (UCHAR*)szLcl},
	};
	MRPROHEAD_S stHead = {
		MR_PRO_FLAG,
		MR_PRO_1,
		MR_PRO_CTRL,
		MR_PRO_0,
		MR_PRO_Y,
		MR_PRO_0,
		MR_PRO_0,
		MR_PRO_0,
		MR_MR_ADP,
		MR_DEF_SUCC,
		MR_PRO_0,
		MR_PRO_0,
	};

	struct option stOptions[] = {
		{"max", 1, 0, 'm'},
		{"sum", 0, 0, 's'},
		{"print", 0, 0, 'p'},
		{"debug", 0, 0, 'd'},
		{"help", 0, 0, 'h'},
		{0, 0, 0, 0},
	};

	while (true) {
		iReturn = getopt_long(argc, argv, "m:spdh", stOptions, &iIndex);
		if (iReturn < 0) {
			break;
		}
		switch (iReturn) {
			case 'm':
				uiMax = atoi(optarg);
				break;
			case 's':
				stHead.m_ucAttr |= MR_MR_SUM;
				break;
			case 'p':
				g_bPrint = true;
				break;
			case 'h':
				MrAtHelp(argv[0]);
				return MR_SUCCESS;
			default:
				return MR_SUCCESS;
		}
	}

	CProSend.MrProPutBody(NULL);
	CProSend.MrProInduce(&stHead);
	CProSend.MrProPutTotal(sizeof(stIn) / sizeof(MRPROKV_S));
	CProSend.MrProPutLength(MrProKvOutput(stIn, CProSend.MrProGetTotal(), &uszSend[MR_PRO_HEAD], MR_PACKET - MR_PRO_HEAD));
	CProSend.MrProCreate(uszSend, MR_PACKET);

	CTime.MrTimeOn();

	for (; uiCount < uiMax; uiCount ++) {
		if (true == g_bPrint) {
			MR_PRINT("Count[%d]:\n", uiCount);
		}
		if (MR_SUCCESS == CUdp.MrNetInitSend(MR_AT_ADDR, MR_AT_PORT)) {
			CUdp.MrNetSetTimeOut(MR_TIMEOUT, 0);
			if (CUdp.MrNetSend(uszSend, MR_PRO_HEAD + CProSend.MrProGetLength()) > 0) {
				memset(szKey, '\0', MR_VALUE);
				memset(uszRecv, '\0', MR_PACKET);
				if ((iLength = CUdp.MrNetRecv(uszRecv, MR_PACKET)) > 0) {
					if (true == g_bPrint) {
						MR_PRINT("Response:\n");
					}
					CProRecv.MrProReset();
					CProRecv.MrProParse(uszRecv, iLength);
					if (true == g_bPrint) {
						CProRecv.MrProDisplay();
					}
					if (CProRecv.MrProGetLength() > 0) {
						MrProKvInput(&stOut, CProRecv.MrProGetTotal(), CProRecv.MrProGetBody(), CProRecv.MrProGetLength());
						memcpy(szKey, stOut.m_puszValue + MR_URL, stOut.m_iLength - MR_URL);
						MrAtResult((char*)stOut.m_puszValue, szKey);
					}
					else {
						MR_ERROR("Failed to MrProGetLength\n");
					}
				}
				else {
					MR_ERROR("Failed to MrNetRecv\n");
				}
			}
			else {
				MR_ERROR("Failed to MrNetSend\n");
			}
			CUdp.MrNetClose();
		}
	}

	CTime.MrTimeOff();
	CTime.MrTimeDisplay();

	return MR_SUCCESS;
}

void MrAtHelp(char* pszProc)
{
	MR_PRINT("Usage: %s [options]\n", pszProc);
	MR_PRINT("Valid options are:\n");
	MR_PRINT("[0] -m or --max:        set max.\n");
	MR_PRINT("[0] -s or --sum:        set sum.\n");
	MR_PRINT("[0] -p or --print:      set print.\n");
	MR_PRINT("[0] -h or --help:       show help.\n");

	return;
}

void MrAtResult(char* pszUrl, char* pszKey)
{
	CMrTcp CTcp;
	int iLength = 0;
	CMrPro CProSend;
	CMrPro CProRecv;
	char szVal[MR_PACKET] = { 0 };
	UCHAR uszSend[MR_PACKET] = { 0 };
	UCHAR uszRecv[MR_PACKET] = { 0 };
	MRPROKV_S stRecv[MR_MR_VAL];
	MRPROKV_S stSend = {MR_MR_KEY, strlen(pszKey), (UCHAR*)pszKey};
	MRPROHEAD_S stHead = {
		MR_PRO_FLAG,
		MR_PRO_1,
		MR_PRO_DATA,
		MR_PRO_0,
		MR_PRO_Y,
		MR_PRO_0,
		MR_DEF_EHUB,
		MR_DEF_SELT,
		MR_PRO_0,
		MR_DEF_SUCC,
		MR_MR_KEY,
		MR_PRO_0,
	};

	CProSend.MrProPutBody(NULL);
	CProSend.MrProInduce(&stHead);
	CProSend.MrProPutLength(MrProKvOutput(&stSend, MR_MR_KEY, &uszSend[MR_PRO_HEAD], MR_PACKET - MR_PRO_HEAD));
	CProSend.MrProCreate(uszSend, MR_PACKET);

	if (MR_SUCCESS == CTcp.MrTcpConnect(pszUrl)) {
		CTcp.MrTcpSetLinger(0);
		CTcp.MrTcpSetNoDelay();
		if (CTcp.MrNetSend(uszSend, MR_PRO_HEAD + CProSend.MrProGetLength()) > 0) {
			memset(uszRecv, '\0', MR_PACKET);
			if (MR_PRO_HEAD == CTcp.MrNetRecv(uszRecv, MR_PRO_HEAD)) {
				if (true == g_bPrint) {
					MR_PRINT("GetResult:\n");
				}
				CProRecv.MrProReset();
				CProRecv.MrProParse(uszRecv, MR_PRO_HEAD);
				if (true == g_bPrint) {
					CProRecv.MrProDisplay();
				}
				if (CProRecv.MrProGetLength() > 0) {
					iLength = CTcp.MrNetRecvAll(&uszRecv[MR_PRO_HEAD], CProRecv.MrProGetLength());
					if (iLength == CProRecv.MrProGetLength()) {
						memset(stRecv, '\0', sizeof(MRPROKV_S) * MR_MR_VAL);
						MrProKvInput(stRecv, MR_MR_VAL, &uszRecv[MR_PRO_HEAD], iLength);
						memcpy(szVal, stRecv[1].m_puszValue, stRecv[1].m_iLength);
						if (true == g_bPrint) {
							MR_PRINT("Key: %s\n", pszKey);
							MR_PRINT("%s\n", szVal);
						}
					}
					else {
						MR_ERROR("Failed to MrNetRecv\n");
					}
				}
				else {
					if (true == g_bPrint) {
						MR_PRINT("Key[%s] not found\n", pszKey);
					}
				}
			}
			else {
				MR_ERROR("Failed to MrNetRecv\n");
			}
		}
		else {
			MR_ERROR("Failed to MrNetSend\n");
		}
		CTcp.MrNetClose();
	}
	else {
		MR_ERROR("Failed to MrTcpConnect\n");
	}

	return;
}
